package com.icreate.storm.bolt;

import com.icreate.storm.methods.indicators.Indicator;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.*;

/**
 * Created by hadoop on 17-5-2.
 */
public class IndicatorBolt extends BaseBasicBolt {
    private Indicator indicator;


    public IndicatorBolt(Indicator indicator) {
        this.indicator = indicator;
    }


    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(indicator.getFields()));
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {

        List<Object> fields = (List<Object>) tuple.getValueByField("fields");
        for (Object field : fields) {
            Values values = indicator.getValues(field);
            if (values != null)
                basicOutputCollector.emit(values);
        }

//        String rows = tuple.getStringByField("fields");
//        String[] strArray = rows.split(LINE_SEPARATOR);
//        for (int i = 0; i < strArray.length; i++) {
//            Values values = indicator.getValues(strArray[i]);
//            if (values != null)
//                basicOutputCollector.emit(values);
//        }
    }

    public void cleanup() {
    }
}
